package net.bwie.realtime.warehouse.ADS;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.ADS
 * @Author: zhangleqing
 * @CreateTime: 2025-09-02  21:56
 * @Description: TODO dwd_refund_reason(退款原因汇总表)
 * 根据【原因类型】和【退款原因】分组统计【退款人数 流失至竞店人数 退款子订单数 订单退款率 退款金额 金额退款率】
 * @Version: 1.0
 */
public class Four {
    private static TableEnvironment tableEnv;

    public static void main(String[] args) {
        // 1 开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2 读取数据
        readTable(tableEnv);

        // 3 计算数据
        Table realTable = handle(tableEnv);

        realTable.execute().print();

        // 4 映射表创建
//        createView(tableEnv);

        // 5 输出数据
//        saveToSink(tableEnv, realTable);
    }

    private static void saveToSink(TableEnvironment tableEnv, Table realTable) {
        tableEnv.createTemporaryView("reason", realTable);
        tableEnv.executeSql(
                ""
        ).print();
    }

    private static void createView(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                ""
        );
    }

/*
根据【原因类型】和【退款原因】分组统计【退款人数 流失至竞店人数 退款子订单数 订单退款率 退款金额 金额退款率】
select
    *
from dwd_refund_reason

 */

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                "select * from dwd_refund_reason limit 10"
        );
        return table;
    }

    private static void readTable(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE dwd_refund_reason (\n" +
                        //（订单id）
                        "order_id STRING ,\n" +
                        // （消费者id）
                        "user_id STRING ,\n" +
                        // (商品ID)
                        "product_id STRING ,\n" +
                        // (订单总金额)
                        "order_amount DOUBLE ,\n" +
                        //(订单支付时间)
                        "pay_time TIMESTAMP(3) ,\n" +

                        // （退款单号）
                        "refund_id STRING ,\n" +
                        // (实际退款金额)
                        "refund_amount DOUBLE ,\n" +

                        // （消费者填写的退款原因）
                        "user_fill_reason STRING ,\n" +

                        // （消费者行为id）
                        "behavior_id STRING ,\n" +
                        // （行为类型）
                        "behavior_type STRING, \n" +
                        // "PRIMARY KEY (order_id) NOT ENFORCED\n"+
                        "WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND\n" +
                        ")WITH(\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'dwd_refund_reason',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'dwd_refund_reason',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
    }

    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
